package hadoop.mr06;


import hadoop.unit.GlobalConfiguration;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;



public class Counter_driver extends Configured implements Tool {
    private static Configuration conf = GlobalConfiguration.getCon();

    @Override
    public int run(String[] args) throws Exception {
        // 判断是否有输入路径和输出路径
        if (args.length < 2) {
            System.err.println("NO TASK");
            System.exit(2); // 输出2为异常退出
        }
        // 构建Job作业的示例 -> 参数(配置对象,Job名称)
        Job job = Job.getInstance(conf, "WordCount_Job");

        // 设置运行主类
        job.setJarByClass(Counter_driver.class);

        // 设置本次程序的mapper和reducer类
        job.setMapperClass(Counter_mapper.class);
        job.setReducerClass(Counter_reduce.class);

        // 设置reduce任务个数 == 设置文件分区个数
//        job.setNumReduceTasks(3);

        // 指定mapper阶段的输出的 key value 数据类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 指定Reducer阶段的输出的 key value 数据类型 也是最终的输出的数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 配置本次任务的输入输出路径
        // todo--默认组件 1、TextInputFormat  2、TextOutputFormat
//       文件路径写死的方法(不推荐) FileInputFormat.setInputPaths(job,new Path("文件路径"));
        Path input_Path = new Path(args[0]); // 输入路径--> hdfs://master:1920/hadoop/Test/WordCount/word.txt
        FileInputFormat.setInputPaths(job, input_Path); // args[0] 文件输入数据路径

        Path output_Path = new Path(args[1]);  // 输出路径--> hdfs://master:1920/hadoop/Test/WordCount/wordResult
        FileOutputFormat.setOutputPath(job, output_Path); // args[1] 文件输出数据路径
        // 如果文件输出路径存在则删除
        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(output_Path)) {
            fs.delete(output_Path, true);
        }
        return job.waitForCompletion(true) ? 0 : 2; // 0为正常退出,2为异常退出
    }

    // 运行任务
    public static void main(String[] args) throws Exception {
        int run = ToolRunner.run(conf, new Counter_driver(), args);
        System.exit(run);
    }
}

